package com.zeyuan.rpcclient.netty.MultiThreadWorker;


import io.netty.buffer.ByteBuf;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class MultiThreadMulWorkerServer {
    public static void main(String[] args) throws IOException {
        //selector boss处理accept，其他worker处理r/w
        Thread.currentThread().setName("boss");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        Selector boss = Selector.open();
        SelectionKey bossKey = ssc.register(boss, 0, null);
        bossKey.interestOps(SelectionKey.OP_ACCEPT);
        ssc.bind(new InetSocketAddress("localhost",8080));
        Worker[] workers=new Worker[Runtime.getRuntime().availableProcessors()];
        for (int i=0;i<workers.length;i++){
            workers[i]=new Worker("worker-"+i);
        }
        AtomicInteger index=new AtomicInteger();
        //创建固定数量的worker并初始化
        Worker work=new Worker("worker-8");

        while (true){
            boss.select();
            Iterator<SelectionKey> iterator = boss.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey next = iterator.next();
                iterator.remove();
                if (next.isAcceptable()){
                    SocketChannel sc = ssc.accept(); //如果有连接事件，则进行连接
                    sc.configureBlocking(false);
//                    work.register(sc);
                    //连接的时候将channel关联到worker这个selector上
                    //问题： 这里跟work.register()里的run方法里面的worker.select()有冲突，因为run方法是线程a，这里worker是线程b，两个是同一个selector，如果线程a中select()阻塞，会影响这里的selector，阻塞结束这里才能执行
//                    sc.register(work.worker,SelectionKey.OP_READ,null); //运行在boss线程中，也就是当前main线程
                    workers[index.getAndIncrement()%workers.length].register(sc); //round robin


                }
            }
        }

    }

    //worker用于处理r/w事件
    static class Worker implements Runnable{
        private Thread thread;
        private Selector worker;
        private String name;
        private volatile boolean start=false; //记录是否初始化
        private ConcurrentLinkedQueue<Runnable> queue=new ConcurrentLinkedQueue<>();

        public Worker(String name) {
            this.name = name;
        }
        //初始化线程和selector
        public void register(SocketChannel sc) throws IOException {
            if (!start){
                thread=new Thread(this,name);
                thread.start();
                worker=Selector.open();
                start=true;
            }
            queue.add(()->{
                try {
                    sc.register(worker,SelectionKey.OP_READ,null);
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                }
            });
            worker.wakeup(); //唤醒select方法先进行注册，注册完成后因为不是事件唤醒的，所以run里面whiletrue又回到selec方法阻塞，然后就变成事件唤醒而且可以监听读事件了
        }

        @Override
        public void run() { //黏包，写数据过多等问题这里不处理了
            while (true){
                try {
                    worker.select(); //这里如果在线程a中阻塞住，会影响外面使用这个selector的地方进行register
                    Runnable poll = queue.poll();
                    if (poll!=null){
                        //这样会导致如果select一直不出现事件，一直阻塞，期间发生的未注册的事件无法监听，所以用了wakeup方法对selector进行唤醒
                        poll.run(); //这样就保证了register中selector和run方法中sworker.select()执行顺序和执行线程的一致，防止出现worker.select()阻塞后，其他线程对这个worker的selector的register注册一直阻塞
                    }
                    //如果有新事件发生
                    Iterator<SelectionKey> iterator = worker.selectedKeys().iterator();
                    while (iterator.hasNext()){
                        SelectionKey next = iterator.next();
                        iterator.remove();
                        if (next.isReadable()){
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            SocketChannel channel = (SocketChannel) next.channel();
                            channel.read(buffer);
                            buffer.flip();
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
